package com.atguigu.tingshu.album.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.common.util.AuthContextHolder;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

	@Autowired
	private AlbumInfoMapper albumInfoMapper;

	@Autowired
	private AlbumAttributeValueService albumAttributeValueService;

	@Autowired
	private AlbumStatMapper albumStatMapper;

	@Autowired
	private KafkaService kafkaService;

	@Autowired
	private RedisTemplate redisTemplate;
	@Override
	public void saveAlbumInfo(AlbumInfoVo albumInfoVo) {

		Long userId = AuthContextHolder.getUserId();
		AlbumInfo albumInfo = new AlbumInfo();
		albumInfo.setUserId(userId);
		BeanUtils.copyProperties(albumInfoVo, albumInfo);
		albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
		String payType = albumInfo.getPayType();
		if(!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(payType))
		{
			albumInfo.setTracksForFree(5);
		}
		albumInfoMapper.insert(albumInfo);

		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();

		 List<AlbumAttributeValue>  albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
					AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
					albumAttributeValue.setAlbumId(albumInfo.getId());
					albumAttributeValue.setAttributeId(albumAttributeValueVo.getAttributeId());
					albumAttributeValue.setValueId(albumAttributeValueVo.getValueId());
					return  albumAttributeValue;
				}).collect(Collectors.toList());

		albumAttributeValueService.saveBatch(albumAttributeValueList);


		//初始化统计数据
		this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_PLAY);
		this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_SUBSCRIBE);
		this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_BROWSE);
		this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_COMMENT);

		if("1".equals(albumInfo.getIsOpen())){
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, String.valueOf(albumInfo.getId()));
		}
	}

	@Override
	public IPage<AlbumListVo> findUserAlbumPage(Integer pageNo, Integer pageSize, AlbumInfoQuery albumInfoQuery) {

		Long userId = AuthContextHolder.getUserId();
		albumInfoQuery.setUserId(userId);
		IPage<AlbumListVo> page = new Page<>(pageNo, pageSize);
		IPage<AlbumListVo> result = albumInfoMapper.findUserAlbumPage(page,albumInfoQuery);

		return result;
	}

	@Transactional
	@Override
	public void removeAlbumInfoById(Long id) {
		albumInfoMapper.deleteById(id);

		LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>();
		albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId, id);
		albumStatMapper.delete(albumStatLambdaQueryWrapper);


		LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper  = new LambdaQueryWrapper<>();
		albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
		albumAttributeValueService.remove(albumAttributeValueLambdaQueryWrapper);

		kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER, String.valueOf(id));
	}

	@Override
	public AlbumInfo getAlbumInfoById(Long id) {

		String redisKey = RedisConstant.ALBUM_INFO_PREFIX + id;
		String albumInfoJson= (String) redisTemplate.opsForValue().get(redisKey);

		if (!StringUtils.isEmpty(albumInfoJson)){
			AlbumInfo albumInfo = JSON.parseObject(albumInfoJson, AlbumInfo.class);
			log.info("从redis中查询专辑数据成功。。。。。");
		return  albumInfo;
		}

		AlbumInfo albumInfo = getAlbumInfoFromDB(id);
		redisTemplate.opsForValue().set(redisKey,JSON.toJSONString(albumInfo),7, TimeUnit.DAYS);
		log.info("从数据库中查询专辑数据成功。。。。。");
		return albumInfo;
	}

	public AlbumInfo getAlbumInfoFromDB(Long id) {

		AlbumInfo albumInfo = albumInfoMapper.selectById(id);

		if (albumInfo != null){
			LambdaQueryWrapper<AlbumAttributeValue> lambdaQueryWrapper = new LambdaQueryWrapper<>();
			lambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId,id );
			albumInfo.setAlbumAttributeValueVoList(albumAttributeValueService.list(lambdaQueryWrapper));


		}

		return albumInfo;
	}

	@Override
	public void updateAlbumInfo(Long id, AlbumInfoVo albumInfoVo) {

		AlbumInfo albumInfo = albumInfoMapper.selectById(id);
		BeanUtils.copyProperties(albumInfoVo,albumInfo);
		albumInfoMapper.updateById(albumInfo);

		LambdaQueryWrapper<AlbumAttributeValue> lambdaQueryWrapper = new LambdaQueryWrapper<>();
		lambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId,id);
		albumAttributeValueService.remove(lambdaQueryWrapper);

		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
			AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
			albumAttributeValue.setAlbumId(id);
			albumAttributeValue.setAttributeId(albumAttributeValue.getAttributeId());
			albumAttributeValue.setValueId(albumAttributeValue.getValueId());
			return albumAttributeValue;

		}).collect(Collectors.toList());

		albumAttributeValueService.saveBatch(albumAttributeValueList);

		if("1".equals(albumInfo.getIsOpen())){
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, String.valueOf(albumInfo.getId()));
		}else {
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER, String.valueOf(albumInfo.getId()));
		}
	}

	@Override
	public List<AlbumInfo> findUserAllAlbumList() {
		Long userId = AuthContextHolder.getUserId();
	LambdaQueryWrapper<AlbumInfo> albumInfoLambdaQueryWrapper = new LambdaQueryWrapper<>();
	albumInfoLambdaQueryWrapper.eq(AlbumInfo::getUserId,userId);
	albumInfoLambdaQueryWrapper.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle);
	List<AlbumInfo> albumInfoList =  albumInfoMapper.selectList(albumInfoLambdaQueryWrapper);
		return albumInfoList;
	}

	@Override
	public AlbumStatVo findAlbumStatByAlbumId(Long albumId) {

		return albumStatMapper.findAlbumStatByAlbumId(albumId);
	}

	public  void saveAlbumStat(Long albumId , String statType) {
		AlbumStat albumStat = new AlbumStat();
		albumStat.setAlbumId(albumId);
		albumStat.setStatType(statType);
		albumStat.setStatNum(0);

		albumStatMapper.insert(albumStat);
	}
}
